package top.zxybird.yun.rpc.api;

import java.lang.reflect.ParameterizedType;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.common.base.Optional;
import com.google.common.primitives.Bytes;

import top.zxybird.yun.common.util.BeanUtils;
import top.zxybird.yun.serialization.thrift.Object2TType;
import top.zxybird.yun.serialization.thrift.TType2Object;
import top.zxybird.yun.serialization.thrift.ThriftBinaryDeserializer;
import top.zxybird.yun.serialization.thrift.ThriftBinarySerializer;
import top.zxybird.yun.serialization.thrift.model.TMessage;
import top.zxybird.yun.serialization.thrift.model.TType;
import top.zxybird.yun.transport.api.TException;
import top.zxybird.yun.transport.api.TTransport;
import top.zxybird.yun.transport.api.TTransportException;
import top.zxybird.yun.transport.thrift.TSocket;

public class ThriftBinaryInvoker extends AbstractInvoker {

  /**
   * seqid生成器
   */
  private final AtomicInteger seqidGenerator = new AtomicInteger(0);

  /**
   * 传输层
   */
  private final TTransport transport;

  private final Class interfaceCls;

  private final Map<String, Class> returnClassMap;

  public ThriftBinaryInvoker(URL url, Class interfaceCls) {
    super(url);
    // 协议要和服务端一致
    this.transport = new TSocket(getServerIp(), getServerPort(), getTimeout());
    try {
      this.transport.open();
    } catch (TTransportException e) {
      throw new RuntimeException("Can not open transport!");
    }
    this.interfaceCls = interfaceCls;
    returnClassMap = BeanUtils.getInterfaceReturnClass(interfaceCls);

  }

  @Override
  @SuppressWarnings("unchecked")
  public Result invoke(Invocation invocation) throws TException {

    // TODO 反射的知识点
    // Class<T> entityClass = (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass())
    // .getActualTypeArguments()[0];

    // 1. Invocation 转 TMessage
    TMessage message = new TMessage();
    message.setType(TMessage.Type.CALL.getType());
    message.setSeqid(seqidGenerator.incrementAndGet());
    message.setName(invocation.getMethod().getName());

    for (Object obj : invocation.getArguments()) {
      message.addField(Object2TType.transform(obj));
    }

    List<Byte> byteList = Optional
        .fromNullable(ThriftBinarySerializer.findByTypeWithDefault(message.getClass()).getFunction().apply(message))
        .or(Collections.emptyList());

    transport.write(Bytes.toArray(byteList));
    transport.flush();

    // 3. 解析返回的数据

    TMessage message_rcv = ThriftBinaryDeserializer.readTMessage(transport);

    // 只返回一个type
    TType _return = message_rcv.getAllField().get(0);
    Object returnObj = TType2Object.transform(_return, returnClassMap.get(invocation.getMethod().getName()));

    // 4. 返回结果
    Result result = new Result();
    result.set(returnObj);

    return result;
  }

}
